package com.boarsoft.boar.batch.service;

import com.boarsoft.bean.ReplyInfo;
import com.boarsoft.boar.batch.biz.BatchExecutionBiz;
import com.boarsoft.boar.batch.entity.BatchExecution;
import com.boarsoft.common.Util;
import com.boarsoft.common.util.RandomUtil;
import com.boarsoft.hibernate.dao.SimpleDao;
import com.boarsoft.rpc.core.RpcContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.transaction.Transactional;
import java.util.List;

/**
 * 批量执行指令服务实现
 *
 * @author tao.Wang57
 * @date 2024-01-08
 */
public class BatchCmdServiceImpl implements BatchCmdService {

    private final static Logger log = LoggerFactory.getLogger(com.boarsoft.boar.batch.service.BatchCmdServiceImpl.class);

    @Autowired
    private BatchExecutionBiz batchExecutionBiz;

    @Autowired
    private SimpleDao simpleDao;

    @Autowired
    private BatchExecCtrlService batchExecCtrlService;

    @Autowired
    private BatchBrokerService batchBrokerService;

    private static final String APP_NAME = "application";

    /**
     * 暂停
     * 步骤：
     * 1、收到暂停请求后，先将当前节点及其子节点的状态全部更新为停止中（6）。
     * 2、暂停指令下发。
     * 3、收到上报方暂停完成的操作后将状态更新为暂停状态（7）。此步骤由产品上报完成后执行
     * <p>
     * STATUS_STOPPING = 6
     * STATUS_STOPPED = 7
     *
     * @return
     */
    @Override
    @Transactional
    public ReplyInfo<Object> pause(String id) {
        //需要暂停的顶级节点
        BatchExecution be = batchExecutionBiz.get(id);
        if (be.getStatus() == BatchExecution.STATUS_SUCCESS) {
            log.error("当前节点已经为完成状态，无法执行暂停操作！id:{},batchId:{},status:{}", be.getId(), be.getBatchId(), be.getStatus());
            return ReplyInfo.success("当前节点已经为完成状态，无法执行暂停操作！");
        }
        //将所有子节点的执行记录包括当前节点的状态全部更新为暂停中状态
        String updSql = " update BatchExecution set tStatus = "
                + BatchExecution.STATUS_STOPPING + " where vPath like '%" + id + "%' and tStatus <> 3 ";
        simpleDao.executeUpdate(updSql);
        //下发暂停状态至批量执行节点
        String addrHql = " from BatchExecution where 1=1 and vPath like '%" + id + "%' and tStatus <> 3 ";
        List<BatchExecution> list = simpleDao.list(addrHql, BatchExecution.class);
        sendCmdStatus(list, BatchExecution.STATUS_STOPPED, "PAUSE");
        return ReplyInfo.SUCCESS;
    }

    @Override
    @Transactional
    public ReplyInfo<Object> pause(String[] ids) {
        for (String id : ids) {
            this.pause(id);
        }
        return ReplyInfo.SUCCESS;
    }

    /**
     * 取消
     * <p>
     * STATUS_CANCELING = 10
     * STATUS_CANCELED = 11
     *
     * @param id
     * @return
     */
    @Override
    @Transactional
    public ReplyInfo<Object> cancle(String id) {
        //需要中断的顶级节点
        BatchExecution be = batchExecutionBiz.get(id);
        if (be.getStatus() == BatchExecution.STATUS_SUCCESS) {
            log.error("当前节点已经为完成状态，无法执行取消操作！id:{},batchId:{},status:{}", be.getId(), be.getBatchId(), be.getStatus());
            return ReplyInfo.success("当前节点已经为完成状态，无法执行取消操作！");
        }
        //将所有子节点的执行记录包括当前节点的状态全部更新为停止状态
        String updSql = " update BatchExecution set tStatus = "
                + BatchExecution.STATUS_CANCELING + " where vPath like '%" + id + "%' and tStatus <> 3 ";
        simpleDao.executeUpdate(updSql);

        //下发停止状态至批量执行节点
        String addrHql = " from BatchExecution where 1=1 and path like '%" + id + "%' and status <> 3 ";
        List<BatchExecution> list = simpleDao.list(addrHql, BatchExecution.class);
        sendCmdStatus(list, BatchExecution.STATUS_CANCELED, "CANCEL");
        return ReplyInfo.SUCCESS;
    }

    @Override
    @Transactional
    public ReplyInfo<Object> cancle(String[] ids) {
        for (String id : ids) {
            this.cancle(id);
        }
        return ReplyInfo.SUCCESS;
    }

    /**
     * 指令发送
     *
     * @param list   需要发送指令的记录监息
     * @param status 需要发送的状态
     * @param cmd    发送的指令名称
     */
    private void sendCmdStatus(List<BatchExecution> list, short status, String cmd) {
        for (BatchExecution b : list) {
            if (Util.strIsNotEmpty(b.getAddr())) {
                try {
                    RpcContext.specify2(b.getAddr());
                    log.info(" send {} to : {} start, batchId:{}", cmd, b.getAddr(), b.getBatchId());
                    batchExecCtrlService.setStatus(b.getBatchId(), status);
                    log.info(" send {} to : {} end, batchId:{}", cmd, b.getAddr(), b.getBatchId());
                    RpcContext.specify2(null);
                } catch (Exception e) {
                    log.error(cmd + " 时发生异常  addr:{} , batchId:{}", b.getAddr(), b.getBatchId());
                    b.setStatus(BatchExecution.STATUS_FAILED);
                    b.setError(cmd + " 时，服务节点失联");
                    batchExecutionBiz.update(b);
                }
            }
        }
    }

    /**
     * 续跑
     * <p>
     * TYPE_RESUME = 1
     *
     * @param id
     * @return
     */
    @Override
    @Transactional
    public ReplyInfo<Object> resume(String id, String appName) {
        log.info("发起 节点id：{} 的续跑调用", id);
        //查询需要续跑的顶级节点及其子节点
        String hql = " from BatchExecution where 1=1 and vId = '" + id + "' and tStatus <> 3 ";
        BatchExecution b = simpleDao.find(hql, BatchExecution.class);
        //新增一条续跑记录
        b.setId(RandomUtil.randomUUID());
        b.setType(BatchExecution.TYPE_RESUME);
        b.setNo((short) (b.getNo() + (short) 1));
        RpcContext.putAtt(APP_NAME, appName);
        try{
            batchBrokerService.resume(b);
        }catch (Exception e){
            log.error("调用续跑时发生异常.", e);
            b.setError("调用续跑时发生异常：" + e);
            b.setStatus(BatchExecution.STATUS_FAILED);
        } finally {
            RpcContext.removeAtt(APP_NAME);
            simpleDao.save(b);
        }
        return ReplyInfo.SUCCESS;
    }

    /**
     * 重跑
     * <p>
     * TYPE_REDO = 2
     *
     * @param id
     * @return
     */
    @Override
    @Transactional
    public ReplyInfo<Object> redo(String id, String appName) {
        log.info("发起 节点id：{} 的重跑调用", id);
        //查询需要重跑的顶级节点及其子节点
        String hql = " from BatchExecution where 1=1 and vId = '" + id + "'";
        BatchExecution b = simpleDao.find(hql, BatchExecution.class);
        //新增一条续跑记录
        b.setId(RandomUtil.randomUUID());
        b.setType(BatchExecution.TYPE_REDO);
        b.setNo((short) (b.getNo() + (short) 1));
        RpcContext.putAtt(APP_NAME, appName);
        try{
            batchBrokerService.redo(b);
        }catch (Exception e){
            log.error("调用重跑时发生异常.", e);
            b.setError("调用重跑时发生异常：" + e);
            b.setStatus(BatchExecution.STATUS_FAILED);
            e.printStackTrace();
        } finally {
            RpcContext.removeAtt(APP_NAME);
            simpleDao.save(b);
        }
        return ReplyInfo.SUCCESS;
    }
}
